package com.hjzd.ai.smartpractice.listener;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.LongSerializationPolicy;
import com.google.gson.reflect.TypeToken;
import com.hjzd.ai.smartpractice.constant.LLMRole;
import com.hjzd.ai.smartpractice.constant.ServiceConstant;
import com.hjzd.ai.smartpractice.entity.DTO.chat.ChatCompletionMessage;
import com.hjzd.ai.smartpractice.entity.DTO.chat.ChatCompletionResponse;
import com.hjzd.ai.smartpractice.service.RedisService;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.protocol.HTTP;
import org.apache.http.util.EntityUtils;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.repository.init.ResourceReader;
import reactor.core.Disposable;
import reactor.core.publisher.FluxSink;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

public class OpenAISubscriber implements Subscriber<String>, Disposable {

    private static final Gson GSON = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss").setLongSerializationPolicy(LongSerializationPolicy.STRING).disableHtmlEscaping().create();
    private static final Logger logger = LoggerFactory.getLogger(OpenAISubscriber.class);

	private final FluxSink<String> emitter;
	private Subscription subscription;
	private final List<String> contentList = new ArrayList<>();
	private String uniqueKey;
	private RedisService redisService;
	private String chatType;

    public OpenAISubscriber(FluxSink<String> emitter, String uniqueKey, RedisService redisService, String chatType) {
        this.emitter = emitter;
		this.uniqueKey = uniqueKey;
		this.redisService = redisService;
		this.chatType = chatType;
    }

    @Override
    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1);
    }

    @Override
    public void onNext(String data) {
        logger.info("OpenAI返回数据：{}", data);
        if ("[DONE]".equals(data)) {
			String contentAll = StringUtils.join(contentList, "");
            logger.info("OpenAI返回数据结束了, contentAll: {}", contentAll);

			ChatCompletionResponse response = new ChatCompletionResponse();
			response.setFinish_reason("stop");
			subscription.request(1);
			if (chatType.equals("chat")) {
				// 将对话完整的存到redis中
				String mockChatInfoStr = redisService.get(String.format(ServiceConstant.MOCK_CHAT_UNIQUE_KEY, uniqueKey));
				List<ChatCompletionMessage> messageList = GSON.fromJson(mockChatInfoStr, new TypeToken<List<ChatCompletionMessage>>() {
				}.getType());
				ChatCompletionMessage message = new ChatCompletionMessage();
				message.setRole(LLMRole.ASSISTANT.getRole());
				message.setContent(contentAll);
				messageList.add(message);
				redisService.set(String.format(ServiceConstant.MOCK_CHAT_UNIQUE_KEY, uniqueKey), GSON.toJson(messageList), 7, TimeUnit.DAYS);
				String ttsAudioUrl = tts(uniqueKey, contentAll);
				response.setReplyAudioUrl(ttsAudioUrl);
			} else if (chatType.equals("suggestion")) {
				redisService.set("MOCK_CHAT_AI_SUGGESTION_" + uniqueKey, contentAll, 7, TimeUnit.DAYS);
			}

            emitter.next(GSON.toJson(response));
            emitter.complete();
        } else {
            ChatCompletionResponse response = GSON.fromJson(data, ChatCompletionResponse.class);
            response.setFinish_reason("unFinish");
            emitter.next(GSON.toJson(response));
            subscription.request(1);
            String content = response.getChoices().get(0).getDelta().getContent();
            if (StringUtils.isNotBlank(content)) {
                contentList.add(content);
            }
        }
    }

	static class CustomerInfo {
		private String name;	// 姓名
		private String gender;	// 性别
		private String address;	// 家庭地址
		private Integer overDueDays;	// 逾期天数
		private String owedAmount;	// 欠款金额
		private String job;		// 工作单位

		public String getName() {
			return name;
		}

		public void setName(String name) {
			this.name = name;
		}

		public String getGender() {
			return gender;
		}

		public void setGender(String gender) {
			this.gender = gender;
		}

		public String getAddress() {
			return address;
		}

		public void setAddress(String address) {
			this.address = address;
		}

		public Integer getOverDueDays() {
			return overDueDays;
		}

		public void setOverDueDays(Integer overDueDays) {
			this.overDueDays = overDueDays;
		}

		public String getOwedAmount() {
			return owedAmount;
		}

		public void setOwedAmount(String owedAmount) {
			this.owedAmount = owedAmount;
		}

		public String getJob() {
			return job;
		}

		public void setJob(String job) {
			this.job = job;
		}
	}

	public String readFileFromResources(String fileName) {

		try {
			ClassPathResource classPathResource = new ClassPathResource(fileName);
			try (InputStream inputStream = classPathResource.getInputStream();
				 BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream))) {
				return reader.lines().collect(Collectors.joining(System.lineSeparator()));
			}
		} catch (Exception ex) {
			return null;
		}

	}

	private String tts(String uniqueKey, String text) {

		String type = null;
		String customerJson = redisService.get(String.format(ServiceConstant.MOCK_CUSTOMER_UNIQUE_KEY, uniqueKey));
		CustomerInfo customerInfo = GSON.fromJson(customerJson, CustomerInfo.class);
		String reference_text= "哦我之前都没注意，我的妈呀，我当时还以为他们是就是找人替的呢。";
		String reference_audio = readFileFromResources("reference_audio.txt");
//		if (customerInfo.getGender().equals("男")) {
//			reference_text = "穿越了好多光年然后去寻找人类另一个栖息地，反正整个电影的特效非常的精彩";
//			reference_audio = "/mnt/data/wav/prompt_wavs/ID0002.wav";
//		} else {
//			reference_text = "这个是我节假日或者就是，嗯外地游客经常会访问的一个景点。";
//			reference_audio = "/mnt/data/wav/prompt_wavs/ID0001.wav";
//		}

		int index = uniqueKey.hashCode() % 2;
		if (index == 0) {
			logger.info(">>> TTS=wenjing");
			reference_text = "相关费用的话，会按照您的还款金额变动，具体的您可以自主查询。那为了避免您产生更多的一个违约金罚息，请您在今晚八点之前还款，好吧。";
			reference_audio = readFileFromResources("reference_audio_wenjing.txt");
		} else {
			logger.info(">>> TTS=puhong");
			reference_text = "您目前的话如果说您家里人，能帮你一起想办法，那你自己也不需要去还一个两万六千八。您只要去还一个两万一千四百五，我们公司帮您还一个七千一百五，那这样子的话这张卡就直接帮你，注销掉就好了";
			reference_audio = readFileFromResources("reference_audio_puhong.txt");
		}

//		String type = redisService.get(ServiceConstant.MOCK_TTS_TYPE);
//		if (StringUtils.isBlank(type)) {
//			reference_text = "哦我之前都没注意，我的妈呀，我当时还以为他们是就是找人替的呢。";
//			reference_audio = readFileFromResources("reference_audio.txt");
//		} else {
//			if (type.equals("xulu")) {
//				reference_text = "先生你好，华夏银行信用卡委托方和君纵达工号一二三四。您华夏银行信用卡目前逾期严重，已经进入银行不良账户名单，现在由总行移交到户籍地，进行了风险调查及信息采集工作。";
//				reference_audio = readFileFromResources("reference_audio_xulu.txt");
//			} else if (type.equals("wenjing")) {
//
//			} else if (type.equals("puhong")) {
//
//			}
//		}




		Map<String, Object> params = new HashMap<>();
		params.put("text", text);
		params.put("reference_text", reference_text);
		params.put("reference_audio", reference_audio);
		params.put("max_new_tokens", 0);
		params.put("chunk_length", 150);
		params.put("top_p", 0.7);
		params.put("repetition_penalty", 1.5);
		params.put("temperature", 0.7);
		params.put("speaker", null);
		params.put("emotion", null);
		params.put("format", "wav");
		params.put("streaming", false);
		params.put("ref_json", "ref_data.json");
		params.put("ref_base", "ref_data");

		try (CloseableHttpClient httpClient = HttpClients.createDefault()) {
			// 创建HttpPost对象
//			Map<String, Object> params = new HashMap<>();
//			params.put("speaker", 2);
//			params.put("content", text);
			HttpPost post = new HttpPost("http://10.252.7.232:8002/v1/invoke");
			post.setHeader(HTTP.CONTENT_TYPE, "application/json; charset=utf-8");
			// 设置请求体
			post.setEntity(new StringEntity(GSON.toJson(params), "UTF-8"));
			// 执行POST请求
			try (CloseableHttpResponse response = httpClient.execute(post)) {
				// 获取响应实体
				HttpEntity entity = response.getEntity();
				if (entity != null) {
					// 输入流获取内容
					String tempFileName = System.currentTimeMillis() + ".wav";
					File file = new File(tempFileName);
					InputStream inputStream = entity.getContent();
					try (OutputStream outputStream = new FileOutputStream(file)) {
						int bytesRead;
						byte[] buffer = new byte[1024];
						while ((bytesRead = inputStream.read(buffer)) != -1) {
							outputStream.write(buffer, 0, bytesRead);
						}
					}
					// 确保实体内容完全消耗
					EntityUtils.consume(entity);
					String audioUrl = saveVoiceFile(uniqueKey, null, new FileInputStream(file), tempFileName);
					file.delete();
					return "https://demo.wangchaotech.com/smart_practice_audio/" + audioUrl;
				}
			}
		} catch (Exception ex) {
			logger.error("TTS转写错误, text={}, 错误信息=", text, ex);
		}
		return null;
	}

	/**
	 * 保存文件并返回可访问的URL
	 * @param uniqueKey 唯一标识
	 * @param role 角色
	 * @param fileStream 文件
	 * @return 可访问的文件URL
	 */
	public String saveVoiceFile(String uniqueKey, String role, InputStream fileStream,String originalFileName) {

		// 提取文件后缀
		String fileExtension = originalFileName.substring(originalFileName.lastIndexOf("."));
		// 相对路径，确保与Nginx的配置相匹配
		String relativePath = Paths.get("/root/servers/smart-practice", "voice", uniqueKey).toString();
		File savePath = new File(relativePath);
		if (!savePath.exists()) {
			savePath.mkdirs();
		}
		// 构建完整文件名
		String newFilename =  System.currentTimeMillis() + fileExtension;
		File targetFile = new File(savePath, newFilename);

		// 将InputStream的内容写入文件
		try (FileOutputStream out = new FileOutputStream(targetFile)) {
			byte[] buffer = new byte[1024];
			int bytesRead;
			while ((bytesRead = fileStream.read(buffer)) != -1) {
				out.write(buffer, 0, bytesRead);
			}
		} catch (IOException e) {
			logger.error("Error saving file: {}", e.getMessage(), e);
			return "Error saving file: " + e.getMessage();
		}
		// 返回修改后的可访问URL
		return uniqueKey + "/" + targetFile.getName();
	}

    @Override
    public void onError(Throwable t) {
		System.out.println("OpenAI返回数据异常：" + t.getMessage());
        logger.error("OpenAI返回数据异常：{}", t.getMessage());
        ChatCompletionResponse response = new ChatCompletionResponse();
        response.setFinish_reason("error");
        emitter.next(GSON.toJson(response));
        emitter.complete();
    }

    @Override
    public void onComplete() {
		System.out.println("OpenAI返回数据完成");
        logger.info("OpenAI返回数据完成");
        emitter.complete();
    }

    @Override
    public void dispose() {
		System.out.println("OpenAI返回数据关闭");
        logger.warn("OpenAI返回数据关闭");
        emitter.complete();
    }
}